package net.bwie.flink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 基于Flink编写流式计算：交易订单事实统计
 * @author xuanyu
 * @date 2025/10/15
 */
public class _03StreamJsonOrderRichFunction {

	/**
	 * 交易订单：{"uid": "u101", "order_amount": 98.9}
	 */
	public static void main(String[] args) throws Exception{
		// 1-执行环境env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1) ;

		// 2-数据源source
		DataStreamSource<String> stream = env.socketTextStream("node101", 9999);

		// 3-数据转换transformation

		// 3-1.过滤脏数据
		SingleOutputStreamOperator<String> stream1 = stream.filter(
			new RichFilterFunction<String>() {

				// 初始化方法，在处理数据之前，准备工作
				@Override
				public void open(Configuration parameters) throws Exception {
					super.open(parameters);
				}

				@Override
				public boolean filter(String value) throws Exception {
					try{
						JSON.parseObject(value);
						return true;
					}catch (Exception e){
						return false;
					}
				}

				// 收尾方法，在数据处理结束之后，清理工作
				@Override
				public void close() throws Exception {
					super.close();
				}
			}
		);

		// 3-2.提取字段值
		SingleOutputStreamOperator<Tuple2<String, Double>> stream2 = stream1.process(
			new ProcessFunction<String, Tuple2<String, Double>>() {
				@Override
				public void processElement(String value, Context ctx, Collector<Tuple2<String, Double>> out) throws Exception {
					/**
					 * value 表示流中每条数据，ctx 表示上下文对象，进行很多高级操作，out表示数据输出
					 */
					// 解析json
					JSONObject jsonObject = JSON.parseObject(value);
					// 提起值
					Double orderAmount = jsonObject.getDouble("order_amount");
					// 二元组
					Tuple2<String, Double> tuple = Tuple2.of("销售额", orderAmount);
					// 输出
					out.collect(tuple);
				}
			}
		);

		// 3-3. 分组
		KeyedStream<Tuple2<String, Double>, String> stream3 = stream2.keyBy(
			new KeySelector<Tuple2<String, Double>, String>() {
				@Override
				public String getKey(Tuple2<String, Double> value) throws Exception {
					return value.f0;
				}
			}
		);

		// 3-4. 求和
		SingleOutputStreamOperator<Tuple2<String, Double>> stream4 = stream3.sum(1);

		// 4-数据终端sink
		stream4.print();

		// 5-触发执行execute
		env.execute("FlinkStreamWordCount") ;
	}

}
